package com.hhf.rrd.usage.window;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;

import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * 多窗口应用
 *
 * @author huanghaifeng15
 * @date 2022/2/15 10:42
 **/
public class MoreWindowApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // 设置watermark
        // ExecutionConfig config = env.getConfig();
        // config.setAutoWatermarkInterval(5L);

        DataStreamSource<String> source = env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (true){
                    ctx.collect("flink");
                }
            }

            @Override
            public void cancel() {

            }
        });

        // 设置 EventTime
        SingleOutputStreamOperator<String> watermark = source.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((SerializableTimestampAssigner<String>) (element, recordTimestamp) -> System.currentTimeMillis()));

        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = watermark.map(new RichMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0);
        tuple2StringKeyedStream.print("map --> ");

        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator1 = tuple2StringKeyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(1L)))
                // .window(TumblingProcessingTimeWindows.of(Time.seconds(2L)))
                .sum(1);
        streamOperator1.print("1111111 --> ");

        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator2 = streamOperator1.map(new RichMapFunction<Tuple2<String, Integer>, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                return value;
            }
        }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                // .window(TumblingEventTimeWindows.of(Time.seconds(1L)))
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1L)))
                .sum(1);
        streamOperator2.print("2222222 --> ");

        env.execute();
    }
}
